package com.citusdata.elven.feeder;

import com.citusdata.elven.datastore.BdbDatastore;
import com.citusdata.elven.feeder.CarDataFeeder;
import com.citusdata.elven.feeder.LinearRoadCallable;
import com.citusdata.elven.feeder.Log4jHelper;
import com.citusdata.elven.linearRoad.protobuf.LinearRoad.*;
import com.citusdata.elven.linearRoad.LinearRoadService;
import com.google.protobuf.Message;
import org.apache.log4j.Logger;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;


public class ConcurrentDataFeeder {
  static { Log4jHelper.setApplicationName(ConcurrentDataFeeder.class); } 
  private static final Logger logger = Logger.getLogger(ConcurrentDataFeeder.class);

  public static void printQueuedResponses(CompletionService completionService, 
                                          int requestCount, PrintWriter output) { 
    try {
      for (int r = 0; r < requestCount; ++r) {
        Future future = completionService.take();
        Message response = (Message) future.get();
        
        String textResponse = CarDataFeeder.serializeResponse(response);
        if (textResponse != null) {
          output.println(textResponse);
        }
      }
    } catch (InterruptedException ex) {
      logger.warn("Exception when reading queued responses: " + ex);
    } catch (ExecutionException ex) {
      logger.warn("Exception when reading queued responses: " + ex);
    }
  }


  public static void main(String[] args) throws IOException {
    if (args.length < 1) {
      System.err.println("Usage: ConcurrentDataFeeder cardata_file " + 
                         "[output_file] [database_dir] [thread_count]");
      System.exit(-1);
    }

    BufferedReader  inputStream = null;
    PrintWriter     outputStream = null;
    BdbDatastore    datastore = null;
    ExecutorService executor  = null;

    try {
      String inputFileName  = args[0];
      String outputFileName = "/var/tmp/notifications.out";
      if (args.length >= 2) {
        outputFileName = args[1];
      }
      inputStream =  new BufferedReader(new FileReader(inputFileName));
      outputStream = new PrintWriter(new FileWriter(outputFileName));

      if (args.length >= 3) {
        // If the user specified a database directory, initialize the data store.
        String bdbDirectory = args[2];
        datastore = new BdbDatastore(bdbDirectory);
      }

      int threadCount = 5;
      if (args.length == 4) {
        threadCount = Integer.parseInt(args[3]);
      }
      executor = Executors.newFixedThreadPool(threadCount);
      CompletionService completions = new ExecutorCompletionService(executor);
      LinearRoadService linearRoad = new LinearRoadService(datastore);

      int requestCount   = 0;
      int previousMinute = 1;
      String inputLine;

      while ((inputLine = inputStream.readLine()) != null) {
        Message request = CarDataFeeder.parseRequest(inputLine);

        int currentMinute = CarDataFeeder.getRequestSemiMinute(inputLine);
        if (currentMinute > previousMinute) {
          printQueuedResponses(completions, requestCount, outputStream);

          requestCount = 0;
          previousMinute = currentMinute;
        }

        requestCount++;
        completions.submit(new LinearRoadCallable(linearRoad, request));
      }

      // Print the remaining responses in the queue
      printQueuedResponses(completions, requestCount, outputStream);
      
    } finally {
      if (executor != null) {
        executor.shutdown();
      }
      if (datastore != null) {
        datastore.close();
      }

      if (inputStream != null) {
        inputStream.close();
      }
      if (outputStream != null) {
        outputStream.close();
      }
    }
  }
}
